package com.chencong.online.function;

import com.chencong.online.bean.AppPromotionCountBean;
import com.chencong.online.utils.TimestampTransformUtil;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;

/**
 * @program: user-behavior-analysis-online
 * @ClassName AppCountProcessWindowFunc
 * @description:将聚合后的分析数据按规则输出
 * @author: chencong
 * @create: 2021-12-29 10:42
 **/
public class AppCountProcessWindowFunc extends ProcessWindowFunction<Long, AppPromotionCountBean, Tuple, TimeWindow> {


    @Override
    public void process(Tuple tuple, ProcessWindowFunction<Long, AppPromotionCountBean, Tuple, TimeWindow>.Context context, Iterable<Long> elements, Collector<AppPromotionCountBean> out) throws Exception {
        String channel = tuple.getField(0);
        String behavior = tuple.getField(1);
        long end = context.window().getEnd();
        LocalDateTime windowEnd = TimestampTransformUtil.milliTimestampToLocalDateTime(end);
        Long count = elements.iterator().next();

        out.collect(new AppPromotionCountBean(channel, behavior, windowEnd, count));

    }
}
